package com.lovecaa.batch;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 创建时间： 2023年10月22日 18:30:00
 *
 * @author Mr.Tao
 * @since 1.0.0
 */
public class BatchMessageProducer {
    public static void main(String[] args) {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("BatchMessageProducer");
            producer.setNamesrvAddr("192.168.192.110:9876");
            producer.start();
            List<Message> messages = new ArrayList<>();
            for (int i = 0; i < 100000; i++) {
                String topic = "Test2";
                String tag = "TagA";
                byte[] msg = ("BatchMessageProducer_" + i).getBytes(StandardCharsets.UTF_8);
                // 使用list把要发送的消息收集起来
                Message message = new Message(topic, tag, msg);
                messages.add(message);
            }
            // 批量分割进行发送
            SplitMessageIterator iterator = new SplitMessageIterator(messages);
            int index = 0;
            while (iterator.hasNext()) {
                producer.send(iterator.next());
                System.out.println("第" + index++ + "次消息发送成功");
            }
            producer.shutdown();
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
    }

    static class SplitMessageIterator implements Iterator<List<Message>> {

        /**
         * 需要分割的消息列表
         */
        private final List<Message> sourceMessages;

        /**
         * 单次分割的最大大小(字节)，这里设置的是3M
         */
        private static final int MAX_SIZE = 1024 * 1024 * 3;

        /**
         * 当前的下标
         */
        private int currentIndex = 0;

        public SplitMessageIterator(List<Message> sourceMessages) {
            this.sourceMessages = sourceMessages;
        }

        @Override
        public boolean hasNext() {
            return currentIndex < sourceMessages.size() - 1;
        }

        @Override
        public List<Message> next() {
            List<Message> result = new ArrayList<>();
            // 本次分割的总大小
            int totalSize = 0;
            for (int i = currentIndex; i < sourceMessages.size(); i++) {
                Message message = sourceMessages.get(i);
                // 消息主要变化的部分是 body+topic+自定义属性 ，所以重点记录这些大小
                int msgSize = message.getBody().length + message.getTopic().getBytes(StandardCharsets.UTF_8).length;
                // 统计自定义属性的大小
                Map<String, String> properties = message.getProperties();
                for (Map.Entry<String, String> entry : properties.entrySet()) {
                    msgSize += entry.getKey().getBytes(StandardCharsets.UTF_8).length;
                    msgSize += entry.getValue().getBytes(StandardCharsets.UTF_8).length;
                }
                result.add(message);
                // 这里的40表示消息里面其他的信息占用的字节数
                totalSize += msgSize + 40;
                currentIndex = i;
                // 如果本次消息的大小大于了分割的大小，就跳出分割开始发送消息
                if (totalSize >= MAX_SIZE) {
                    break;
                }
            }
            return result;
        }
    }
}
